package com.spzx.order.receiver;

import com.rabbitmq.client.Channel;
import com.spzx.common.rabbit.constant.MqConst;
import com.spzx.order.configure.DeadLetterMqConfig;
import com.spzx.order.configure.DelayedMqConfig;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class TestReceiver {

    @Autowired
    RedisTemplate redisTemplate;


    /**
     * 方式二：通过注解声明交换机和队列，以及绑定。
     */
    @RabbitListener(
            bindings = {
                    @QueueBinding(
                            value = @Queue(name = MqConst.QUEUE_TEST, durable = "true", exclusive = "false", autoDelete = "false"), //默认，持久化，不排他，不自动删除
                            exchange = @Exchange(name = MqConst.EXCHANGE_TEST, type = ExchangeTypes.DIRECT, durable = "true", autoDelete = "false", internal = "false"), //默认，持久化，不自动删除，不是内部
                            key = {
                                MqConst.ROUTING_TEST
                            }
                    )
            }
    )
    public void recevier(String msg, Message message, Channel channel) {
        log.info("msg="+msg);

        //(Body:'hello' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8,
        // contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false,
        // receivedExchange=spzx.test, receivedRoutingKey=spzx.test, deliveryTag=1,
        // consumerTag=amq.ctag-lF5UH9r9h_7sWClatRWOLA, consumerQueue=spzx.test])
        log.info("message="+message);
        Long messageId = message.getMessageProperties().getDeliveryTag(); //消息唯一标识，类似id
        try {
            channel.basicAck(messageId,false); // multiple=true,一次确认多个消息；  false一次只确认一个消息。
        } catch (IOException e) {
            try {
                channel.basicNack(messageId,false,true); // requeue=true回到原来队列。requeue=false 丢弃或进入死信队列。
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
        }
    }




    /**
     * 监听确认消息
     * @param message
     */
    @SneakyThrows //抛出异常
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = MqConst.EXCHANGE_TEST, durable = "true"),
            value = @Queue(value = MqConst.QUEUE_CONFIRM, durable = "true"),
            key = MqConst.ROUTING_CONFIRM
    ))
    public void confirm(String content, Message message, Channel channel) {
        log.info("接收确认消息：{}", content);

        // false 确认一个消息，true 批量确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }



    /**
     * 监听延迟消息
     * @param msg
     * @param message
     * @param channel
     */
    @SneakyThrows
    @RabbitListener(queues = {DeadLetterMqConfig.queue_dead_2}) //监听死信队列。
    public void getDeadLetterMsg(String msg, Message message, Channel channel) {
        log.info("死信消费者：{}", msg);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }



    /**
     * 监听延迟消息(解决幂等性)
     * @param msg
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = { DelayedMqConfig.queue_delay_1 })
    public void getDeadLetterMsg2222(String msg, Message message, Channel channel)
            throws IOException {
        log.info("TestReceiver-消息：{}", msg);

        String lockKey = "mq:"+msg;
        Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockKey, "", 200, TimeUnit.SECONDS);
        //没有设置锁成功，说明消息已经被处理，无需重复处理，直接确认即可。
        if(!flag){
            log.info("TestReceiver-没有获取锁，直接确认该消息。");
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            return;
        }
        //如果设置锁成功,说明第一次处理消息。TODO 执行业务逻辑代码。
        log.info("TestReceiver-处理消息幂等性，执行业务处理。first");
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}
